package com.zengjianjun.hadoop.demo.client;

import cn.hutool.core.util.IdUtil;
import com.zengjianjun.hadoop.demo.HadoopDemoApplication;
import com.zengjianjun.hadoop.demo.config.HadoopConfig;
import com.zengjianjun.hadoop.demo.config.props.HadoopProperties;
import com.zengjianjun.hadoop.demo.executor.JobTypeEnum;
import com.zengjianjun.hadoop.demo.executor.mapReduce.WordCountMap;
import com.zengjianjun.hadoop.demo.executor.mapReduce.WordCountReduce;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @author zengjianjun
 */
@Slf4j
@Component
public class MapReduceClient {

    @Resource
    private HadoopClient hadoopClient;
    @Resource
    private HadoopProperties hadoopProperties;

    /**
     * 单词统计，统计某个单词出现的次数
     *
     * @param jobTypeEnum {@link JobTypeEnum} 任务类型
     * @param inputPath   文件路径，如果是文件家路径，则默认分词该文件夹下所有的文件
     */
    public void wordCount(JobTypeEnum jobTypeEnum, String inputPath) {
        // 输出目录 = output/当前Job_uuid,如果输出路径存在则删除，保证每次都是最新的
        String outputPath = hadoopProperties.getFullOutputPath(jobTypeEnum.name() + "_" + IdUtil.randomUUID());
        hadoopClient.rmdir(outputPath);
        this.getWordCountJobsConf(jobTypeEnum.name(), inputPath, outputPath);
    }

    /**
     * 获取单词统计的配置信息
     *
     * @param jobName    名称
     * @param inputPath  待统计文件地址
     * @param outputPath 输出地址
     */
    public void getWordCountJobsConf(String jobName, String inputPath, String outputPath) {
        try {
            // 创建一个 Job
            Job job = Job.getInstance(new HadoopConfig().getConfiguration(hadoopProperties), jobName);
            // 设置运行的主类
            job.setJarByClass(HadoopDemoApplication.class);
            // 设置 Mapper 和 Reducer
            job.setMapperClass(WordCountMap.class);
            job.setCombinerClass(WordCountReduce.class);
            job.setReducerClass(WordCountReduce.class);
            // 设置 Mapper 输出 key 和 value 的类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
            // 设置 Reducer 输出 key 和 value 的类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
            // 设置作业输入文件和输出文件的路径
            FileInputFormat.addInputPath(job, new Path(inputPath));
            FileOutputFormat.setOutputPath(job, new Path(outputPath));
            // 将作业提交到群集并等待它完成，参数设置为 true 代表打印显示对应的进度
            boolean result = job.waitForCompletion(true);
            log.info("MapReduceClient.getWordCountJobsConf result: {}", result);
        } catch (Exception e) {
            log.error("MapReduceClient.getWordCountJobsConf error: ", e);
        }
    }
}
